package org.fuys.etp.core.pool;

import lombok.extern.slf4j.Slf4j;
import org.fuys.etp.core.metrics.EtpMetricsHandlerChain;
import org.fuys.etp.core.model.event.EtpReportEvent;
import org.fuys.etp.core.model.req.TpStrategyChangeReq;
import org.fuys.etp.core.model.res.TpConfigUpdateRes;
import org.fuys.etp.core.model.res.TpStatusResult;
import org.fuys.etp.core.model.res.TpStrategyChangeRes;
import org.fuys.etp.core.model.type.EtpTypeEnum;
import org.fuys.etp.core.model.vo.TpBaseConfigVO;
import org.fuys.etp.core.model.vo.TpConfigVO;
import org.fuys.etp.core.model.vo.TpDynamicConfigVO;
import org.fuys.etp.core.model.vo.TpStatusVO;
import org.fuys.etp.core.pool.algorithm.PIDController;
import org.fuys.etp.core.pool.algorithm.TaskCountPredictor;
import org.fuys.etp.core.pool.queue.ResizeableLinkedBlockingQueue;
import org.springframework.context.ApplicationEventPublisher;

import java.lang.reflect.InvocationTargetException;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @projectName: fuys-etp-final
 * @package: org.fuys.etp.core.pool
 * @className: EtpThreadPoolExecutor
 * @author: WangJing
 * @description: 弹性线程池
 * @date: 2025/2/10 21:18
 * @version: 1.0
 */
@Slf4j
public class EtpThreadPoolExecutor extends ThreadPoolExecutor {
    public EtpThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    public EtpThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
    }

    public EtpThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    public EtpThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    //限制用配置 对于阶段切换类型的线程池而言 限制用配置是实际用配置的下一阶段 而对于自适应的线程池 限制用配置只规定最大的资源利用程度
    //实际用配置才是当前的线程池具体配置状态
    private TpConfigVO limitConfig;

    //实际用配置
    private TpConfigVO actualConfig;

    //PID增强对任务趋势的预测 只有此线程池类型为自调整的线程池类型才会存在此对象 否则为null 此举是为了减少配置复杂性
    private PIDController pidController;

    //滑动窗口+移动加权算法预测任务趋势
    private TaskCountPredictor taskCountPredictor;

    //线程池名称
    private String name;

    //线程池类型
    private String type;

    //线程池报警类型
    private String alertType;

    //接管状态
    private volatile AtomicBoolean takeover;

    //上报执行者
    private ScheduledExecutorService reportExecutor;

    //事件发送者
    private ApplicationEventPublisher eventPublisher;

    private EtpMetricsHandlerChain metricsHandlerChain;

    public void initExecutor(String name,String type,String alertType
            ,TpConfigVO limitConfig,TpConfigVO actualConfig,
                             ApplicationEventPublisher eventPublisher,EtpMetricsHandlerChain chain){
        this.eventPublisher = eventPublisher;
        this.metricsHandlerChain = chain;
        this.actualConfig = actualConfig;
        this.limitConfig = limitConfig;
        this.name = name;
        this.type = type;
        this.alertType = alertType;
        this.takeover = new AtomicBoolean(false);
        if(EtpTypeEnum.ADAPTIVE_THREAD_POOLS_EXECUTOR.type().equals(type)){
            //初始化PID和滑动窗口
            this.pidController = new PIDController(actualConfig.getDynamicConfig().getPidTarget(),
                    actualConfig.getDynamicConfig().getPidKp(),actualConfig.getDynamicConfig().getPidKi(),
                    actualConfig.getDynamicConfig().getPidKd());
            this.taskCountPredictor = new TaskCountPredictor(actualConfig.getDynamicConfig().getWindowsSize(),
                    actualConfig.getDynamicConfig().getWindowMinSize(),actualConfig.getDynamicConfig().getWindowWeights());
        }
        reportExecutor = Executors.newScheduledThreadPool(1);
        final ScheduledFuture<?> scheduledFuture = reportExecutor.scheduleAtFixedRate(() -> {
                    try {
                        log.info("Fuys etp:发送自检事件");
                        final TpStatusResult statusResult = this.statusExplore();
                        EtpReportEvent event = new EtpReportEvent(this, statusResult);
                        eventPublisher.publishEvent(event);
                    } catch (Exception e) {
                        e.printStackTrace();
                        log.error("Fuys etp:出现异常");
                    }
                }, actualConfig.getBaseConfig().getSelfCheckInterval().longValue(),
                actualConfig.getBaseConfig().getSelfCheckInterval().longValue(),
                TimeUnit.valueOf(actualConfig.getBaseConfig().getSelfCheckIntervalUnit()));
        log.info("Fuys etp 弹性线程池创建完成 创建参数为:名称:{},类型:{},限制配置:{},实际配置:{}"
                ,name,type,limitConfig,actualConfig);
    }

    public TpStatusResult statusExplore(){
        TpStatusVO statusVO = metricsHandlerChain.collectStatus(this);
        return TpStatusResult.builder()
                .poolType(this.type)
                .poolName(this.name)
                .takeover(this.takeover())
                .configVO(this.actualConfig)
                .limitConfigVO(this.limitConfig)
                .tpStatusVO(statusVO).build();
    }
    
    public String getType(){
        return this.type;
    }
    
    public String getName(){
        return this.name;
    }

    public PIDController getPidController(){
        return this.pidController;
    }

    public TaskCountPredictor getTaskPredictor(){
        return this.taskCountPredictor;
    }

    public TpConfigVO getActualConfig(){
        return this.actualConfig;
    }

    public TpConfigVO getLimitConfig(){
        return this.limitConfig;
    }

    public TpStrategyChangeRes changeStrategy(String changeType,TpConfigVO limitConfig,TpConfigVO actualConfig)
            throws InvocationTargetException, IllegalAccessException, NoSuchMethodException {
        AtomicBoolean change = new AtomicBoolean(true);
        TpConfigVO backupLimit = this.limitConfig;
        TpConfigVO backupActual = this.actualConfig;
        try{
            if(this.type.equals(changeType)){
                //如果是同类型
                this.changeBaseConfig(limitConfig,actualConfig);
            }else{
                //策略类型不同 则根据切换的策略类型进行修改
                if(EtpTypeEnum.ADAPTIVE_THREAD_POOLS_EXECUTOR.type().equals(changeType)){
                    //要切换的类型是动态类型 则赋值检测器
                    final TpDynamicConfigVO dynamicConfig = actualConfig.getDynamicConfig();
                    this.pidController = new PIDController(dynamicConfig.getPidTarget(), dynamicConfig.getPidKp(),
                            dynamicConfig.getPidKi(),dynamicConfig.getPidKd());
                    this.taskCountPredictor = new TaskCountPredictor(dynamicConfig.getWindowsSize(),
                            dynamicConfig.getWindowMinSize(),dynamicConfig.getWindowWeights());
                    this.changeBaseConfig(limitConfig,actualConfig);
                } else if(EtpTypeEnum.STAGE_SWITCH_THREAD_POOL_EXECUTOR.type().equals(changeType)){
                    //由动态切换到阶梯式 则取消检测器
                    this.pidController = null;
                    this.taskCountPredictor = null;
                    this.changeBaseConfig(limitConfig,actualConfig);
                }
                this.type = changeType;
            }
            return TpStrategyChangeRes.ok();
        }catch (Exception e){
            e.printStackTrace();
            change.compareAndSet(true,false);
            return TpStrategyChangeRes.builder().extInfo(e.getMessage())
                    .changeType(changeType).build();
        }finally {
            if(!change.get()){
                //切换时出现异常 回滚
                this.changeBaseConfig(backupLimit,backupActual);
                //如果此方法依然出现异常 则自动回滚失败 需要人工干涉查明原因 异常会被抛出
            }
        }
    }

    public TpConfigUpdateRes updateConfig(TpConfigVO limitConfig,TpConfigVO actualConfig)
            throws InvocationTargetException, IllegalAccessException, NoSuchMethodException {
        AtomicBoolean change = new AtomicBoolean(true);
        TpConfigVO backupLimit = this.limitConfig;
        TpConfigVO backupActual = this.actualConfig;
        try {
            this.changeBaseConfig(limitConfig,actualConfig);
            return TpConfigUpdateRes.normal();
        }catch (Exception e){
            e.printStackTrace();
            change.compareAndSet(true,false);
            return TpConfigUpdateRes.builder().extInfo(e.getMessage()).success(false)
                    .build();
        }finally {
            if(!change.get()){
                //切换时出现异常 回滚
                this.changeBaseConfig(backupLimit,backupActual);
                //如果此方法依然出现异常 则自动回滚失败 需要人工干涉查明原因 异常会被抛出
            }
        }
    }


    public boolean takeover(){
        return takeover.get();
    }

    public void confirmTakeover(){
        takeover.compareAndSet(false, true);
    }

    public void relieveTakeover(){
        takeover.compareAndSet(true, false);
    }

    private void changeBaseConfig(TpConfigVO limitConfig,TpConfigVO actualConfig)
            throws InvocationTargetException, IllegalAccessException, NoSuchMethodException {
        TpBaseConfigVO changeTarget = actualConfig.getBaseConfig();
        final TpBaseConfigVO curConfig = this.actualConfig.getBaseConfig();
        if(!changeTarget.getWorkQueueSize().equals(curConfig.getWorkQueueSize())){
            final BlockingQueue<Runnable> queue = this.getQueue();
            if(queue instanceof LinkedBlockingQueue) {
                ResizeableLinkedBlockingQueue.setCapacity((LinkedBlockingQueue<Runnable>) queue,
                        changeTarget.getWorkQueueSize());
            }
        }
        if(!changeTarget.getPoolCoreSize().equals(curConfig.getPoolCoreSize())){
            this.setCorePoolSize(changeTarget.getPoolCoreSize());
        }
        if(!changeTarget.getPoolMaxSize().equals(curConfig.getPoolMaxSize())){
            this.setMaximumPoolSize(changeTarget.getPoolMaxSize());
        }
        if(!changeTarget.getPoolAliveTimeUnit().equals(curConfig.getPoolAliveTimeUnit())
        || !changeTarget.getPoolKeepAliveTime().equals(curConfig.getPoolKeepAliveTime())){
            this.setKeepAliveTime(changeTarget.getPoolKeepAliveTime(),
                    TimeUnit.valueOf(changeTarget.getPoolAliveTimeUnit().toString()));
        }
        if(!changeTarget.getRejectHandleBean().equals(curConfig.getRejectHandleBean())){
            this.setRejectedExecutionHandler(changeTarget.getRejectHandle());
        }
        this.actualConfig = actualConfig;
        this.limitConfig = limitConfig;
    }

    public String getAlertType() {
        return this.alertType;
    }

    @Override
    public String toString(){
        return "EtpThreadPoolExecutor{" +
                "name='" + name + '\'' +
                ", type='" + type + '\'' +
                ", alertType='" + alertType + '\'' +
                ", takeover=" + takeover +
                ", limitConfig=" + limitConfig +
                ", actualConfig=" + actualConfig+
                ", predictor"+ taskCountPredictor +
                ", pidController"+pidController +"}";
    }
}
